media_pp\elements\sink/
app_sink.rs

1use std::sync::Arc;
2
3use crate::pp_log::{PpLog, pp_info};
4
5use crate::{
6    buffer::MediaBuffer,
7    control::ControlMsg,
8    element::{Element, ElementType, Sink, element_pp_log},
9    error::Result,
10};
11
12/// Terminal sink that hands every buffer (and, optionally, every control
13/// message) to a plain closure instead of requiring a bespoke `struct` +
14/// `Element`/`Sink` impl — the equivalent of GStreamer's `appsink`: the
15/// pipeline's job ends here, and whatever the caller does with the data
16/// (run inference, forward it to a channel, write it out, ...) is none of
17/// this crate's concern.
18///
19/// `FrameCounter`/`PacketCounter` are what a one-off consumer looked
20/// like *before* this existed — this is the general case of the same
21/// pattern, for when a whole new type per use site is more ceremony than
22/// the actual logic warrants:
23///
24/// ```
25/// # use media_pp::{buffer::MediaBuffer, elements::AppSink};
26/// let mut count = 0usize;
27/// let sink = AppSink::new("counter", move |buf: MediaBuffer| {
28///     if matches!(buf, MediaBuffer::Video(_)) {
29///         count += 1;
30///     }
31///     Ok(())
32/// });
33/// ```
34pub struct AppSink<F, C> {
35    pp_log: PpLog,
36    name: Arc<str>,
37    consume: F,
38    control: C,
39}
40
41impl<F> AppSink<F, fn(ControlMsg) -> Result<()>>
42where
43    F: FnMut(MediaBuffer) -> Result<()> + Send + 'static,
44{
45    /// `consume` is the only thing this reacts to — every `ControlMsg`
46    /// (`Pause`/`Resume`/`Stop`/`Seek`) is silently ignored, the same as
47    /// `FrameCounter`/`PacketCounter`. Reach for
48    /// [`AppSink::with_control`] instead if the closure needs to know
49    /// about one of those — e.g. resetting a tracker's history, or a
50    /// batch buffer, on `Seek`, the same way `SwDecoder`/`Pacer` react to
51    /// it internally.
52    pub fn new(name: impl Into<String>, consume: F) -> Self {
53        Self::with_control(name, consume, |_| Ok(()))
54    }
55}
56
57impl<F, C> AppSink<F, C>
58where
59    F: FnMut(MediaBuffer) -> Result<()> + Send + 'static,
60    C: FnMut(ControlMsg) -> Result<()> + Send + 'static,
61{
62    /// Same as [`AppSink::new`], but also hands every [`ControlMsg`] to
63    /// `control` instead of silently dropping it.
64    ///
65    /// ```
66    /// # use media_pp::{control::ControlMsg, elements::AppSink};
67    /// let sink = AppSink::with_control(
68    ///     "detector",
69    ///     |_buf| Ok(()),
70    ///     |msg| {
71    ///         if let ControlMsg::Seek(_) = msg {
72    ///             // e.g. clear a tracker's history here
73    ///         }
74    ///         Ok(())
75    ///     },
76    /// );
77    /// ```
78    pub fn with_control(name: impl Into<String>, consume: F, control: C) -> Self {
79        let name: Arc<str> = name.into().into();
80        let pp_log = element_pp_log(ElementType::AppSink, &name, None);
81        pp_info!(pp_log: &pp_log, "created");
82        Self {
83            name,
84            pp_log,
85            consume,
86            control,
87        }
88    }
89}
90
91impl<F, C> Element for AppSink<F, C>
92where
93    F: FnMut(MediaBuffer) -> Result<()> + Send + 'static,
94    C: FnMut(ControlMsg) -> Result<()> + Send + 'static,
95{
96    fn name(&self) -> Arc<str> {
97        self.name.clone()
98    }
99
100    fn element_type(&self) -> ElementType {
101        ElementType::AppSink
102    }
103
104    fn pp_log(&self) -> &PpLog {
105        &self.pp_log
106    }
107
108    fn pp_log_mut(&mut self) -> &mut PpLog {
109        &mut self.pp_log
110    }
111}
112
113impl<F, C> Sink for AppSink<F, C>
114where
115    F: FnMut(MediaBuffer) -> Result<()> + Send + 'static,
116    C: FnMut(ControlMsg) -> Result<()> + Send + 'static,
117{
118    fn consume(&mut self, buf: MediaBuffer) -> Result<()> {
119        (self.consume)(buf)
120    }
121
122    fn control(&mut self, msg: ControlMsg) -> Result<()> {
123        (self.control)(msg)
124    }
125}
media_pp\elements\sink/app_sink.rs

media_pp\elements\sink/
app_sink.rs
